package com.icarbonx.dap.notification.msgservice.redis

import java.util.concurrent.ConcurrentHashMap

import redis.clients.jedis.exceptions.JedisConnectionException
import redis.clients.jedis.{Jedis, JedisPool, JedisPoolConfig}

import scala.collection.JavaConversions._


object ConnectionPool {
  @transient private lazy val pools: ConcurrentHashMap[RedisEndpoint, JedisPool] =
    new ConcurrentHashMap[RedisEndpoint, JedisPool]()

  def getPool(re: RedisEndpoint): JedisPool = {
    pools.getOrElseUpdate(re, {
      val poolConfig: JedisPoolConfig = new JedisPoolConfig();
      poolConfig.setMaxTotal(250)
      poolConfig.setMaxIdle(32)
      poolConfig.setTestOnBorrow(true)
      poolConfig.setTestOnReturn(true)
      poolConfig.setTestWhileIdle(false)
      poolConfig.setMinEvictableIdleTimeMillis(60000)
      poolConfig.setTimeBetweenEvictionRunsMillis(30000)
      poolConfig.setNumTestsPerEvictionRun(-1)
      new JedisPool(poolConfig, re.host, re.port, re.timeout, re.password)
    }
    )
  }

  def connect(re: RedisEndpoint): Jedis = {
    val pool = getPool(re)
    var sleepTime: Int = 4
    var conn: Jedis = null
    while (conn == null) {
      try {
        conn = pool.getResource
      }
      catch {
        case e: JedisConnectionException if e.getCause.toString.
          contains("ERR max number of clients reached") => {
          if (sleepTime < 500) sleepTime *= 2
          Thread.sleep(sleepTime)
        }
        case e: Exception => throw e
      }
    }
    conn
  }

  def returnResource(re: RedisEndpoint, jedis: Jedis) = {
    val pool = getPool(re)
    pool.returnResource(jedis)
  }

  def close(re: RedisEndpoint): Unit = {
    pools.get(re).close()
  }

  val hook = new Thread {
    override def run = pools.values().foreach(_.close())
  }
  sys.addShutdownHook(hook.run)


}

